package com.xb.test.utils;


import com.xb.test.pool.KafkaPool;
import com.xb.test.pool.KafkaProducerApp;
import org.apache.kafka.clients.producer.KafkaProducer;

import java.util.Properties;

public class KafkaUtils
{





    static Properties props = new Properties();
    static KafkaProducer<String, String> producer = null;
    static KafkaPool<String, String> pool = null;
    static {

        Properties props = new Properties();
        props.put("bootstrap.servers", "58.218.203.30:9092,58.218.201.170:9092,58.218.201.172:9092,58.218.201.167:9092,58.218.203.29:9092");
//        props.put("bootstrap.servers","192.168.1.200:9092,192.168.1.201:9092,192.168.1.202:9092");
        props.put("acks", "-1");
        props.put("retries", 30);
        props.put("compression.type","gzip");
        props.put("groupid","xb1");
        props.put("reconnect.backoff.ms ", 20000);
        props.put("retry.backoff.ms", 20000);
        props.put("batch.size", 20000);
        props.put("linger.ms", 50);
        props.put("buffer.memory", 335544320);
        props.put("key.serializer",
                "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer",
                "org.apache.kafka.common.serialization.StringSerializer");

        pool=new KafkaPool<String, String>(props);


    }

    public static boolean sendMessage(String str, String kafkaTopic){


        try {
            KafkaProducerApp<String, String> producer =pool.borrowProducer();
            producer.send(kafkaTopic,str);
            pool.returnProducer(producer);
        } catch (Exception e) {
            e.printStackTrace();
            return false;
        }

        return true;

    }
}
